package com.xiliu.rocketmq;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author xiliu
 * @description  发送异步消息 demo
 * @date 2022/3/2
 */
public class AsyncProducerDemo {

    public static void main(String[] args) throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer("xiliu_producer_group");
        producer.setNamesrvAddr("42.194.222.32:9876");
        // 发送失败重试次数
        producer.setRetryTimesWhenSendAsyncFailed(0);

        producer.start();

        Message msg = new Message("xiliu_topic","xiliu_async_tag","Hello 我是异步消息".getBytes(RemotingHelper.DEFAULT_CHARSET));

        final CountDownLatch countDownLatch = new CountDownLatch(1);
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("消息状态：" + sendResult.getSendStatus());
                System.out.println("消息id：" + sendResult.getMsgId());
                System.out.println("消息queue：" + sendResult.getMessageQueue());
                System.out.println("消息offset：" + sendResult.getQueueOffset());
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("发送失败！" + throwable);
                throwable.printStackTrace();
            }
        },10000);
        countDownLatch.await(5, TimeUnit.SECONDS);
        producer.shutdown();
    }
}
